package niit.producer

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import java.util.Properties
import scala.util.Random

object producer {

  // 生成student数据
  private def generateStudentInfo(): String = {
    // 生成随机的学生信息数据，班级，姓名，性别，出生日期，学号，学期，是否在籍
    // val randomInt = random.nextInt()
    val classId = 202300 + Random.nextInt(5)
    val name = s"Student${Random.nextInt(100)}"
    val gender = Random.nextInt(2)
    val birthDate = s"${1990 + Random.nextInt(11)}-${1 + Random.nextInt(12)}-${1 + Random.nextInt(28)}"
    val studentId = s"S${1000000000 + Random.nextInt(900000000)}"
    val semester = (1 + Random.nextInt(8)).toString
    val enrollmentStatus = if (Random.nextBoolean()) "L" else "A"

    // 将学生信息数据拼接成字符串并返回，每行数据以\t分隔
    s"$classId\t$name\t$gender\t$birthDate\t$studentId\t$semester\t$enrollmentStatus" //202301	Student60	1	1996-1-26	S1736146197	1	L
  }


  def main(args: Array[String]): Unit = {
    // kafka的配置
    val props = new Properties()
    //指定Kafka集群的地址和端口
    props.put("bootstrap.servers", "192.168.157.128:9092")
    //将键对象序列化为字节数组的类，这里使用StringSerializer
    props.put("key.serializer", classOf[StringSerializer].getName)
    //将值对象序列化为字节数组的类，这里使用StringSerializer
    props.put("value.serializer", classOf[StringSerializer].getName)

    //创建了一个KafkaProducer对象，用于生产消息到Kafka主题，key类型是String，value类型是String
    val producer = new KafkaProducer[String, String](props)

    //while循环
    try {
      while (true) {
        val studentInfo = generateStudentInfo()
        println(studentInfo)
        // 发送学生信息数据到 Kafka 主题
        producer.send(new ProducerRecord[String, String]("stuInfo", studentInfo))
        // 模拟每隔 5 秒发送一次数据
        Thread.sleep(5000)
      }
    }
    //使用try-catch-finally语句块来捕获InterruptedException异常
    catch {
      case _: InterruptedException =>
    } finally {
      producer.close()
    }
  }
}
